package com.atguigu.day02

import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}

import scala.util.Random


object Example5 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val intDataStream = env.addSource(new RichParallelSourceFunction[Int] {
      private var flag = true
      private val random = new Random

      override def run(sourceContext: SourceFunction.SourceContext[Int]): Unit = {
        while (flag) {
          sourceContext.collect(random.nextInt(1000))
          Thread.sleep(500)
        }
      }

      override def cancel(): Unit = {
        flag = false
      }
    })
    //求和
//    intDataStream
//      .keyBy(x=>true)
//      .reduce((x,y)=>x+y)
//      .print()
    //求平均值
//    intDataStream
//      .map(x=>(x,1))
//      .keyBy(x=>true)
//      .reduce((x,y)=>(x._1+y._1,x._2+y._2))
//      .map(x=>x._1/x._2.toDouble)
//      .print()
    //求最大最小值


    env.execute()
  }


}

